package com.fancy.middleware.netty.handler.v3;

import com.fancy.middleware.common.annotation.V3Server;
import com.fancy.middleware.common.beans.Domain;
import com.fancy.middleware.common.beans.DomainWrapper;
import com.fancy.middleware.common.constant.RedisConstant;
import com.fancy.middleware.common.product.FrameDataEventProducer;
import com.fancy.middleware.common.service.SessionService;
import com.fancy.middleware.common.service.redis.RedisOperationService;
import com.fancy.middleware.common.utils.ContextUtils;
import com.fancy.middleware.common.utils.StringUtil;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service("cacheFrameHandler")
@ChannelHandler.Sharable
@V3Server
@Slf4j
public class V3CacheFrameHandler extends ChannelInboundHandlerAdapter {

    @Autowired
    private FrameDataEventProducer<DomainWrapper> frameDataEventProducer;

    @Autowired
    private RedisOperationService redisOperationService;


    /***
     * @Description:
     *  1. 数据存入redis中
     *  2. 将domain type 和 序列号存入 ringbuffer 中
     *  3. 消费者消费时根据 type 和序列号 确定唯一数据   消费者消费时候 需要使用分布式事务锁去处理
     * @param ctx
     * @param msg
     * @Author: fancy
     * @Date: 2023/1/7
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            Domain domain = (Domain) msg;
            SessionService sessionService = ContextUtils.getBeanByType(SessionService.class);
            log.debug("bind Session for SerialNumber {}", domain.getFrameData().getSerialNumber());
            sessionService.bind(domain.getFrameData().getSerialNumber(), ctx.channel());
            // TODO insert into data to redis
            String key = StringUtil.format(RedisConstant.FRAME_DATA_LIST_KEY,
                    String.valueOf(domain.getDomainType()),
                    String.valueOf(domain.getFrameData().getSerialNumber()));
            Object o = redisOperationService.lockAndExpire(RedisConstant.LOCK_WRITER + key, 1000, () -> {
                // TODO 插入数据到 redis list 尾部 消费者从redis list 头部消费
                return redisOperationService.getRedisUtils().lRightPush(key, domain);
            });
            // TODO modify ringbuffer data
            frameDataEventProducer.onData(domain, ctx);
        } catch (Exception e) {
            log.error(e.getMessage());
            // TODO 数据处理失败 需要手动处理帧数据 保证数据不丢失
        }
    }
}
